package cn.itbeien.payment.controller;

import cn.itbeien.payment.message.SimpleMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author beien
 * @date 2024-03-22 13:32
 * Copyright© 2024 beien
 */
@RestController
@Slf4j
public class MessageController {
    /**
     * 通过springcloud StreamBridge对象发送消息到生产者
     */
    @Autowired
    private StreamBridge streamBridge;

    @RequestMapping("sendMsg")
    public void sendMsg(){
        String key = "KEY" + 1;
        //消息头
        Map<String, Object> headers = new HashMap<>();
        headers.put(MessageConst.PROPERTY_KEYS, key);
        headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, 1);
        // 设置延时等级1~10  延迟30秒发送消息
        headers.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 4);
        Message<SimpleMsg> msg = new GenericMessage(new SimpleMsg("Delay RocketMQ " + 1), headers);
        streamBridge.send("producer-out-0", msg);
    }

    /**
     * 延迟消费生产者中的消息
     * @return
     */
    //@Bean
    public Consumer<Message<SimpleMsg>> consumer() {
        return msg -> {
            log.info(Thread.currentThread().getName() + " Consumer Receive New Messages: " + msg.getPayload().getMsg());
        };

    }

}
